package com.yh.csx.bsf.ods;

import com.yh.csx.bsf.core.base.Ref;
import com.yh.csx.bsf.core.thread.ThreadPool;
import com.yh.csx.bsf.core.util.ContextUtils;
import com.yh.csx.bsf.core.util.LogUtils;
import com.yh.csx.bsf.ods.meta.OdsSyncAdapter;
import com.yh.csx.bsf.ods.monitor.CheckSchedule;

import lombok.var;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.scheduling.annotation.EnableScheduling;

@EnableScheduling
@SpringBootApplication
@ComponentScan("com.yh.csx")
public class CanelIncrementStarter {
	@Bean
	public StragtegyProvider getStragtegyProvider() {
		return new StragtegyProvider();
	}

	public static void main(String[] args) {
		try {
			SpringApplication.run(CanelIncrementStarter.class, args);
			if (CommonUtils.getSubscribeDbs().size() == 0) {
				throw new RuntimeException("未配置待同步的订阅库,示范:increment.subscribe.10.19.238.92=csx_b2b_crm");
			}			
			var server = new CanalServerProvider();
			server.start();
			// 暂缓一下
			try {
				Thread.sleep(5 * 1000);
			} catch (Exception e) {
			}
			// 配置新增实例

			Map<String, OdsSyncAdapter> syncers = ContextUtils.getApplicationContext().getBeansOfType(OdsSyncAdapter.class);
			com.yh.csx.bsf.redis.RedisProvider redisProvider = ContextUtils.getBean(com.yh.csx.bsf.redis.RedisProvider.class, false);
			if (Optional.ofNullable(syncers).orElseThrow(RuntimeException::new).isEmpty()) {
				throw new Exception("syncers is null");
			}
			syncers.forEach((key, OdsSyncAdapter) -> {
				LogUtils.info(CanelIncrementStarter.class, "ods","同步器已配置：" + key + ":" + OdsSyncAdapter.getClass().getName());
			});
			CheckSchedule  checktask=ContextUtils.getBean(CheckSchedule.class, false);
			checktask.freshConfigs();
			
		
			StragtegyProvider stragtegyProvider = ContextUtils.getBean(StragtegyProvider.class, true);
			Set<String> servers=new HashSet<String>();
			for (var db : CommonUtils.getSubscribeDbs()) {
				Map<String, String> mapInfo=stragtegyProvider.getDbInfo(db);
				servers.add(mapInfo.get("address"));
				
			}
			Ref<Short> i= new Ref<Short>(Short.valueOf("1"));
			for (String db :servers) {
				LogUtils.info(CanelIncrementStarter.class, "ods", db+"-clientid:"+i.getData()+",canal 增量消费者已启动!");
				i.setData((short)(i.getData().shortValue()+(short)1));
				ThreadPool.System.submit("canal-task", () -> {
					CanalIncrementConsumer test = new CanalIncrementConsumer(syncers, redisProvider, stragtegyProvider);
					test.setAddress("127.0.0.1:11111");
					test.setClientId((short)(1001+i.getData()));
					test.setDebug(true);
					test.setDestination(db);
					test.setFilter(".*\\..*");
					test.setUsername("");
					test.setPassword("");
					test.setBatchSize(stragtegyProvider.getCanalBatchSize());
					test.start();
					
				});
			}
			LogUtils.info(CanelIncrementStarter.class, "ods", "canal 增量消费者已启动!");
		} catch (Exception e) {
			LogUtils.error(CanelIncrementStarter.class, "ods", "严重错误", e);
		}

	}

}
